package cool.webstudy.async.mq.kafka.listener.log.opl;

import com.fasterxml.jackson.databind.ObjectMapper;
import cool.webstudy.async.service.OperationLoggerService;
import cool.webstudy.common.constant.LoggerConstant;
import cool.webstudy.common.model.dto.logger.opl.CreateOperationLogESDTO;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @author 莫振双
 * @date 2024/8/27 17:43
 * @description: 审计日志记录消费者
 */
@Component
public class OperationLoggerESListener {
    @Autowired
    private OperationLoggerService operationLoggerService;

    @KafkaListener(topics = LoggerConstant.OPL_LOG_ES_TOPIC, groupId = LoggerConstant.LOGGER_GROUP_ID)
    public void listener(ConsumerRecord<?, ?> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional message = Optional.ofNullable(record.value());
        if (message.isPresent()) {
            Object msg = message.get();
            String msgStr = (String) msg;
            if (StringUtils.isBlank(msgStr)) {
                return;
            }
            try {
                //反序列化
                CreateOperationLogESDTO dto = new ObjectMapper().readValue(msgStr, CreateOperationLogESDTO.class);
                operationLoggerService.createOperationLogToES(dto);
            } catch (Exception e) {
                //TODO 将日志写入磁盘日志文件中，在UTIL中创建保存日志文件至磁盘的工具类
                return;
            }
        }
    }
}
